package com.ruoyi.stream.server.handler;

import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.common.utils.spring.SpringUtils;
import com.ruoyi.stream.framework.annotations.StreamServiceEndpoint;
import com.ruoyi.stream.framework.core.protocol.SKAProtocolContext;
import com.ruoyi.stream.framework.core.protocol.SKAProtocolHeader;
import com.ruoyi.stream.framework.core.protocol.StreamAliveRequest;
import com.ruoyi.stream.framework.core.protocol.StreamAliveResponse;
import com.ruoyi.stream.framework.enums.IOType;
import com.ruoyi.stream.framework.enums.SKARequestOption;
import com.ruoyi.stream.framework.enums.SKAResponseOption;
import com.ruoyi.stream.framework.model.StreamServiceEndpointHandler;
import com.ruoyi.stream.framework.model.request.RequestProcessor;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import jakarta.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class ServiceHandler extends SimpleChannelInboundHandler<StreamAliveRequest> {
    private static final Map<String,StreamServiceEndpointHandler> handlers = new HashMap<>();
    @Autowired
    List<StreamServiceEndpointHandler> handlerList;

    private final SKAProtocolContext context = new SKAProtocolContext();

    private static ThreadPoolTaskExecutor executor = SpringUtils.getBean("threadPoolTaskExecutor");


    private RequestProcessor processor;

    @PostConstruct
    public void init(){
        for (StreamServiceEndpointHandler handler : handlerList) {
            StreamServiceEndpoint annotation = handler.getClass().getAnnotation(StreamServiceEndpoint.class);
            if (annotation == null|| StringUtils.isEmpty(annotation.uri())){
                continue;
            }
            ServiceHandler.handlers.put(annotation.uri(),handler);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, StreamAliveRequest streamAliveRequest)
            throws Exception {
        executor.execute(()->{
            String option = streamAliveRequest.getOption();
            if (SKARequestOption.REQUEST_RESOURCE.getOptionString().equals(option)){
                String uri = streamAliveRequest.getUri();
                StreamServiceEndpointHandler endpointHandler = handlers.get(uri);
                if (endpointHandler == null){
                    StreamAliveResponse response = new StreamAliveResponse();
                    response.setBody("can not find handler: " + uri );
                    response.setResponseOption(SKAResponseOption.ERROR);
                    response.setToReqId(streamAliveRequest.getReqId());
                    channelHandlerContext.writeAndFlush(response);
                    return;
                }
                context.setOnRequest(streamAliveRequest);
                String type = streamAliveRequest.getHeader().get(SKAProtocolHeader.RESOURCE_TYPE);
                if (StringUtils.isEmpty(type)){
                    StreamAliveResponse response = new StreamAliveResponse();
                    response.setBody("unSupport request");
                    response.setResponseOption(SKAResponseOption.ERROR);
                    response.setToReqId(streamAliveRequest.getReqId());
                    channelHandlerContext.writeAndFlush(response);
                    return;
                }
                RequestProcessor requestProcessor = null;
                for (IOType value : IOType.values()) {
                    if (value.type.equals(type)){
                        try {
                            requestProcessor = value.constructor.newInstance();
                            break;
                        } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
                            throw new RuntimeException(e);
                        }
                    }
                }
                if (requestProcessor == null){
                    StreamAliveResponse response = new StreamAliveResponse();
                    response.setBody("unSupport request type: ");
                    response.setResponseOption(SKAResponseOption.ERROR);
                    response.setToReqId(streamAliveRequest.getReqId());
                    channelHandlerContext.writeAndFlush(response);
                    return;
                }
                requestProcessor.setHandler(endpointHandler);
                requestProcessor.setContext(context);
                requestProcessor.setCtx(channelHandlerContext);
                processor = requestProcessor;
                StreamAliveResponse response = new StreamAliveResponse();
                response.setResponseOption(SKAResponseOption.READY);
                requestProcessor.handleRequestResource(response);
            } else if (SKARequestOption.READY_ACCEPT.getOptionString().equals(option)) {
                if (!checkPre(streamAliveRequest,channelHandlerContext)){
                    return;
                }
                StreamAliveResponse response = new StreamAliveResponse();
                response.setResponseOption(SKAResponseOption.READY);
                response.setToReqId(streamAliveRequest.getReqId());
                processor.handleReady(response);
            } else if (SKARequestOption.REFUSE.getOptionString().equals(option)) {
                if (!checkPre(streamAliveRequest,channelHandlerContext)){
                    return;
                }
                StreamAliveResponse response = new StreamAliveResponse();
                response.setResponseOption(SKAResponseOption.REFUSE);
                response.setToReqId(streamAliveRequest.getReqId());
                processor.handleRefuse(streamAliveRequest,response);
            }
        });
    }

    private boolean checkPre(StreamAliveRequest request, ChannelHandlerContext context){
        if (processor == null){
            StreamAliveResponse response = new StreamAliveResponse();
            response.setBody("require before request ");
            response.setResponseOption(SKAResponseOption.ERROR);
            response.setToReqId(request.getReqId());
            context.writeAndFlush(response);
            return false;
        }
        return true;
    }

}
